<?php
//设置你的连接
$conn_args = array(
    'host' => 'localhost', 
    'port' => '5672', 
    'login' => 'jsapi', 
    'password' => '123456',
    'vhost' => 'applications');
$conn = new AMQPConnection($conn_args);
if ($conn->connect()) {
    echo "Established a connection to the broker \n";
}
else {
    echo "Cannot connect to the broker \n ";
}


//创建channel
$channel = new AMQPChannel($conn);
//创建queue
$queue = new AMQPQueue($channel);
$queue->setName('test.queue.lu');//创建名字
$queue->setFlags(AMQP_DURABLE);
//$queue->declare();
/*
$message = $queue->get(AMQP_AUTOACK);

if($message){
    var_dump(json_decode($message -> getBody(), true));
}
else{
    echo 'message is empty';
}

*/

try {
    while ($envelope = $queue->get()) {
        $message_id = $envelope->getMessageId();
        $message_id = isset($message_id) ? $message_id : null;
        $content_type = $envelope->getContentType();
        $content_type = isset($content_type) ? $content_type : null;
        $delivery_mode = $envelope->getDeliveryMode();
        $priority = $envelope->getPriority();
        $priority = isset($priority) ? $priority : null;
        $timestamp = $envelope->getTimeStamp();
        $timestamp = isset($timestamp) ? $timestamp : null;
        //if (isset($timestamp)) {
        //    $timestamp = GeneralFunctions::formatTimestamp($timestamp, 'Y-m-d H:i:s');
        //}
        $app_id = $envelope->getAppId();
        $app_id = isset($app_id) ? $app_id : null;
        $correlation_id = $envelope->getCorrelationId();
        $correlation_id = isset($correlation_id) ? $correlation_id : null;
        $reply_to = $envelope->getReplyTo();
        $reply_to = isset($reply_to) ? $reply_to : null;
        $content_encoding = $envelope->getContentEncoding();
        $content_encoding = isset($content_encoding) ? $content_encoding : null;
        $expiration = $envelope->getExpiration();
        $expiration = isset($expiration) ? $expiration : null;
        $type = $envelope->getContentType();
        $type = isset($type) ? $type : null;
        $queue_name = $envelope->getRoutingKey();
        $arr_headers = $envelope->getHeaders();
       //$retry_number = isset($arr_headers[GeneralFunctions::RETRY]) ? $arr_headers[GeneralFunctions::RETRY] : 0;
        //$err_code = isset($arr_headers[GeneralFunctions::ERR_NO]) ? $arr_headers[GeneralFunctions::ERR_NO] : 0;
        //$err_msg = isset($arr_headers[GeneralFunctions::ERR_MSG]) ? $arr_headers[GeneralFunctions::ERR_MSG] : '';
        //$recovery_number = isset($arr_headers[GeneralFunctions::RECOVERY]) ? $arr_headers[GeneralFunctions::RECOVERY] : 0;
        //if ($recovery_number == -1) {
        //    $recovery_action_code = Quarantine::DISCARDED;
        //} else {
            //$recovery_action_code = ($recovery_number >= $this->max_recovery || $recovery_number < 0)? Quarantine::NONRECOVERABLE : Quarantine::NOT_SPECIFIED;
        //}
        $headers = isset($arr_headers) ? serialize($arr_headers) : null;
        $payload = $envelope->getBody();
        $payload = isset($payload) ? $payload : null;

        echo '1.'.$message_id.
            ' 2.'.$content_type.
            ' 3.'.$delivery_mode.
            ' 4.'.$priority.
            ' 5.'.$timestamp.
            ' 6.'.$app_id.
            ' 7.'.$correlation_id.
            ' 8.'.$reply_to.
            ' 9.'.$content_encoding.
            ' 10.'.$expiration.
            ' 11.'.$type.
            ' 12.'.$queue_name.
            ' 14.'.$headers.
            ' 15.'.$payload;

        //do work, eg: insert data to DB
        $status = false;

        if ($status) {
            // ACK and drop message if retry more than limit number
            $queue->ack($envelope->getDeliveryTag());
        } else {

            // NACK and requeue message if failed to retry
            $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);            
        }
    }
} catch (Exception $e) {
    echo $e->getMessage();
    //$this->logger->logError('process_queue', $e);
}


$conn -> disconnect();

?>